/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package cn.edu.thu.laud.objectstore.query;

import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.Stack;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.datanucleus.ObjectManager;
import org.datanucleus.exceptions.NucleusException;
import org.datanucleus.metadata.AbstractClassMetaData;
import org.datanucleus.metadata.AbstractMemberMetaData;
import org.datanucleus.query.compiler.CompilationComponent;
import org.datanucleus.query.compiler.QueryCompilation;
import org.datanucleus.query.evaluator.AbstractExpressionEvaluator;
import org.datanucleus.query.expression.CastExpression;
import org.datanucleus.query.expression.CreatorExpression;
import org.datanucleus.query.expression.Expression;
import org.datanucleus.query.expression.InvokeExpression;
import org.datanucleus.query.expression.Literal;
import org.datanucleus.query.expression.ParameterExpression;
import org.datanucleus.query.expression.PrimaryExpression;
import org.datanucleus.query.expression.SubqueryExpression;
import org.datanucleus.query.expression.VariableExpression;
import cn.edu.thu.laud.utils.serde.ConverterContext;
import cn.edu.thu.laud.objectstore.query.CassandraBooleanExpression;
import cn.edu.thu.laud.objectstore.query.CassandraExpression;
import cn.edu.thu.laud.objectstore.query.CassandraFieldExpression;
import cn.edu.thu.laud.objectstore.query.CassandraLiteral;
import cn.edu.thu.laud.objectstore.query.CassandraUnSupportedExpression;
import cn.edu.thu.laud.objectstore.index.IndexContainer;
import cn.edu.thu.laud.objectstore.index.IndexMeta;
import cn.edu.thu.laud.utils.MetaDataUtils;
import org.datanucleus.store.query.Query;
import org.datanucleus.util.NucleusLogger;

/**
 * Class which maps a compiled (generic) query to a Cassandra query.
 * @author zhuoan
 */
public class QueryToCassandraMapper extends AbstractExpressionEvaluator
{
    final ObjectManager om;

    final String candidateAlias;

    final AbstractClassMetaData candidateCmd;

    final Query query;

    final QueryCompilation compilation;

    /** Input parameter values, keyed by the parameter name. Will be null if compiled pre-execution. */
    final Map<?, ?> parameters;

    /** Work Map for keying parameter value for the name, for case where parameters input as positional. */
    Map<String, Object> parameterValueByName = null;

    Map<Integer, String> paramNameByPosition = null;

    /** Positional parameter that we are up to (-1 implies not being used). */
    int positionalParamNumber = -1;

    /** State variable for the component being compiled. */
    CompilationComponent compileComponent;

    /** Whether the filter clause is completely evaluatable in the datastore. */
    boolean filterComplete = true;
       
    CassandraBooleanExpression filterExpr= null;
    
    IndexMeta idxMeta;
    
    int count;
    
    Cassandra.Client client;
    
    private ConverterContext converterContext;
    
    Set<ByteBuffer> disColValues;
    
    ByteBuffer disColName;
    
    Stack<CassandraExpression> stack = new Stack<CassandraExpression>();

    public QueryToCassandraMapper(ConverterContext converterContext, Cassandra.Client client, QueryCompilation compilation, Map<?, ?> parameters, int count, AbstractClassMetaData cmd,
            ObjectManager om, Query q, IndexMeta idxMeta, ByteBuffer disColName, Set<ByteBuffer> disColValues)
    {
        this.om = om;
        this.converterContext = converterContext;
        this.query = q;
        this.client = client;
        this.compilation = compilation;
        this.parameters = parameters;
        this.count = count;
        this.candidateCmd = cmd;
        this.candidateAlias = compilation.getCandidateAlias();
        this.idxMeta = idxMeta;
        this.disColName = disColName;
        this.disColValues = disColValues;
    }

    public boolean isFilterComplete()
    {
        return filterComplete;
    }
    
    public CassandraBooleanExpression getFilterExpression()
    {
        return filterExpr;
    }

    /**
     * Method to compile the WHERE clause of the query
     */
    public void compileFilter()
    {
        if (compilation.getExprFilter() != null)
        {
            try{
	        	compileComponent = CompilationComponent.FILTER;
	            compilation.getExprFilter().evaluate(this);
	            CassandraExpression cassExpr = stack.pop();
                if (cassExpr instanceof CassandraBooleanExpression)
                {
                    this.filterExpr = (CassandraBooleanExpression)cassExpr;
                }
                else
                {
                    NucleusLogger.QUERY.error(">> invalid compilation : filter compiled to " + filterExpr);
                    filterComplete = false;
                }
	            
            }
            catch (Exception e)
            {
                // Impossible to compile all to run in the datastore, so just exit
                filterComplete = false;
                NucleusLogger.QUERY.debug(">> compileFilter caught exception ", e);
            }
            compileComponent = null;
        }
    }

    /* (non-Javadoc)
     * @see org.datanucleus.query.evaluator.AbstractExpressionEvaluator#processOrExpression(org.datanucleus.query.expression.Expression)
     */
    @Override
    protected Object processOrExpression(Expression expr)
    {
    	
    	CassandraExpression right = stack.pop();    	
    	CassandraExpression left =  stack.pop();
    	if(left instanceof CassandraUnSupportedExpression || right instanceof CassandraUnSupportedExpression){
    		stack.push(CassandraUnSupportedExpression.getInstance());
    		filterComplete = false;
    		return null;
    	}
    	else{    		
	    	CassandraBooleanExpression orExpr = new CassandraBooleanExpression((CassandraBooleanExpression)left, (CassandraBooleanExpression)right, Expression.OP_OR);
	        stack.push(orExpr);
	        return orExpr;
    	}
    }

    /* (non-Javadoc)
     * @see org.datanucleus.query.evaluator.AbstractExpressionEvaluator#processAndExpression(org.datanucleus.query.expression.Expression)
     */
    @Override
    protected Object processAndExpression(Expression expr)
    {
    	CassandraExpression right = stack.pop();    	
    	CassandraExpression left =  stack.pop();
    	if(left instanceof CassandraUnSupportedExpression && right instanceof CassandraBooleanExpression){
    		stack.push(right);
    		filterComplete = false;
    		return right;
    	}
    	else if(left instanceof CassandraBooleanExpression && right instanceof CassandraUnSupportedExpression){
    		stack.push(left);
    		filterComplete = false;
    		return left;
    	}
    	else if(left instanceof CassandraUnSupportedExpression && right instanceof CassandraUnSupportedExpression){
    		stack.push(CassandraUnSupportedExpression.getInstance());
    		filterComplete = false;
    		return null;
    	}
    	else{    		
	    	CassandraBooleanExpression andExpr = new CassandraBooleanExpression((CassandraBooleanExpression)left, (CassandraBooleanExpression)right, Expression.OP_AND);
	        stack.push(andExpr);
	        return andExpr;
    	}
    }
    
    /**
     * Method to process the supplied LIKE expression.
     * To be implemented by subclasses.
     * @param expr The expression
     * @return The result
     */
    protected Object processLikeExpression(Expression expr)
    {
    	stack.push(CassandraUnSupportedExpression.getInstance());    	
    	return null;
    }
    
    protected Object processNoteqExpression(Expression expr)
    {
        //same to like
    	stack.push(CassandraUnSupportedExpression.getInstance());
    	return null;
    }
    
    protected Object processIsExpression(Expression expr)
    {
    	stack.push(CassandraUnSupportedExpression.getInstance());
    	return null;
    }
    
    

    @Override
	protected Object processIsnotExpression(Expression expr) {
		// TODO Auto-generated method stub
    	stack.push(CassandraUnSupportedExpression.getInstance());
    	return null;
	}

	@Override
	protected Object processInExpression(Expression expr) {
		// TODO Auto-generated method stub
		stack.push(CassandraUnSupportedExpression.getInstance());
    	return null;
	}
	
	@Override
	protected Object processAddExpression(Expression expr) {
		// TODO Auto-generated method stub
		stack.push(CassandraUnSupportedExpression.getInstance());
    	return null;
	}

	@Override
	protected Object processSubExpression(Expression expr) {
		// TODO Auto-generated method stub
		stack.push(CassandraUnSupportedExpression.getInstance());
    	return null;
	}

	@Override
	protected Object processMulExpression(Expression expr) {
		// TODO Auto-generated method stub
		stack.push(CassandraUnSupportedExpression.getInstance());
    	return null;
	}

	@Override
	protected Object processDivExpression(Expression expr) {
		// TODO Auto-generated method stub
		stack.push(CassandraUnSupportedExpression.getInstance());
    	return null;
	}

	@Override
	protected Object processModExpression(Expression expr) {
		// TODO Auto-generated method stub
		stack.push(CassandraUnSupportedExpression.getInstance());
    	return null;
	}

	@Override
	protected Object processNegExpression(Expression expr) {
		// TODO Auto-generated method stub
		stack.push(CassandraUnSupportedExpression.getInstance());
    	return null;
	}

	@Override
	protected Object processComExpression(Expression expr) {
		// TODO Auto-generated method stub
		stack.push(CassandraUnSupportedExpression.getInstance());
    	return null;
	}

	@Override
	protected Object processNotExpression(Expression expr) {
		// TODO Auto-generated method stub
		stack.push(CassandraUnSupportedExpression.getInstance());
    	return null;
	}

	@Override
	protected Object processDistinctExpression(Expression expr) {
		// TODO Auto-generated method stub
		stack.push(CassandraUnSupportedExpression.getInstance());
    	return null;
	}

	@Override
	protected Object processCreatorExpression(CreatorExpression expr) {
		// TODO Auto-generated method stub
		stack.push(CassandraUnSupportedExpression.getInstance());
    	return null;
	}

	@Override
	protected Object processVariableExpression(VariableExpression expr) {
		// TODO Auto-generated method stub
		stack.push(CassandraUnSupportedExpression.getInstance());
    	return null;
	}

	@Override
	protected Object processSubqueryExpression(SubqueryExpression expr) {
		// TODO Auto-generated method stub
		stack.push(CassandraUnSupportedExpression.getInstance());
    	return null;
	}

	@Override
	protected Object processInvokeExpression(InvokeExpression expr) {
		// TODO Auto-generated method stub
		stack.push(CassandraUnSupportedExpression.getInstance());
    	return null;
	}

	@Override
	protected Object processCastExpression(CastExpression expr) {
		// TODO Auto-generated method stub
		stack.push(CassandraUnSupportedExpression.getInstance());
    	return null;
	}

	/* (non-Javadoc)
     * @see org.datanucleus.query.evaluator.AbstractExpressionEvaluator#processEqExpression(org.datanucleus.query.expression.Expression)
     */
    @Override
    protected Object processEqExpression(Expression expr)
    {        
    	 try{
	    	 Object right = stack.pop();
	         Object left = stack.pop();
	         if (left instanceof CassandraLiteral && right instanceof CassandraFieldExpression)
	         {
	        	 CassandraFieldExpression fieldExpr = (CassandraFieldExpression)right;
	             CassandraLiteral litExpr = (CassandraLiteral)left;
	             IndexContainer container = new IndexContainer(idxMeta.getItem(), fieldExpr.getIndexName());
	             CassandraExpression cassExpr = new CassandraBooleanExpression(client, ByteBufferUtil.string(fieldExpr.getColumnName()),
	                 litExpr.getValue(), Expression.OP_EQ, count, disColName, disColValues, container, idxMeta);
	             stack.push(cassExpr);
	             return cassExpr;
	         }
	         else if (right instanceof CassandraLiteral && left instanceof CassandraFieldExpression)
	         {
	        	 CassandraFieldExpression fieldExpr = (CassandraFieldExpression)left;
	        	 CassandraLiteral litExpr = (CassandraLiteral)right;
	        	 IndexContainer container = new IndexContainer(idxMeta.getItem(), fieldExpr.getIndexName());
	             CassandraExpression cassExpr = new CassandraBooleanExpression(client, ByteBufferUtil.string(fieldExpr.getColumnName()),
	                 litExpr.getValue(), Expression.OP_EQ, count, disColName, disColValues, container, idxMeta);
	             stack.push(cassExpr);
	             return cassExpr;
	         }
    	 }catch (Exception e){
    		 throw new NucleusException(e.getMessage(), e);    		 
    	 }
    	 
    	 stack.push(CassandraUnSupportedExpression.getInstance());
     	 return null;
    }
    
    /* (non-Javadoc)
     * @see org.datanucleus.query.evaluator.AbstractExpressionEvaluator#processGtExpression(org.datanucleus.query.expression.Expression)
     */
    @Override
    protected Object processGtExpression(Expression expr)
    {       
    	try{
	    	Object right = stack.pop();
	        Object left = stack.pop();
	        if (left instanceof CassandraLiteral && right instanceof CassandraFieldExpression)
	        {
	       	    CassandraFieldExpression fieldExpr = (CassandraFieldExpression)right;
	            CassandraLiteral litExpr = (CassandraLiteral)left;
	            IndexContainer container = new IndexContainer(idxMeta.getItem(), fieldExpr.getIndexName());
	            CassandraExpression cassExpr = new CassandraBooleanExpression(client, ByteBufferUtil.string(fieldExpr.getColumnName()),
	            	litExpr.getValue(), Expression.OP_LT, count, disColName, disColValues, container, idxMeta);
	            stack.push(cassExpr);
	            return cassExpr;
	        }
	        else if (right instanceof CassandraLiteral && left instanceof CassandraFieldExpression)
	        {
	       	    CassandraFieldExpression fieldExpr = (CassandraFieldExpression)left;
	       	    CassandraLiteral litExpr = (CassandraLiteral)right;
	       	    IndexContainer container = new IndexContainer(idxMeta.getItem(), fieldExpr.getIndexName());
	            CassandraExpression cassExpr = new CassandraBooleanExpression(client, ByteBufferUtil.string(fieldExpr.getColumnName()),
	            	litExpr.getValue(), Expression.OP_GT, count, disColName, disColValues, container, idxMeta);
	            stack.push(cassExpr);
	            return cassExpr;
	        }
    	} catch (Exception e){
    		throw new NucleusException(e.getMessage(), e);    		
    	}
    	
    	stack.push(CassandraUnSupportedExpression.getInstance());
    	return null;
    }

    /* (non-Javadoc)
     * @see org.datanucleus.query.evaluator.AbstractExpressionEvaluator#processLtExpression(org.datanucleus.query.expression.Expression)
     */
    @Override
    protected Object processLtExpression(Expression expr)
    {        
    	try{
	    	Object right = stack.pop();
	        Object left = stack.pop();
	        if (left instanceof CassandraLiteral && right instanceof CassandraFieldExpression)
	        {
	       	    CassandraFieldExpression fieldExpr = (CassandraFieldExpression)right;
	            CassandraLiteral litExpr = (CassandraLiteral)left;
	            IndexContainer container = new IndexContainer(idxMeta.getItem(), fieldExpr.getIndexName());
	            CassandraExpression cassExpr = new CassandraBooleanExpression(client, ByteBufferUtil.string(fieldExpr.getColumnName()),
	                litExpr.getValue(), Expression.OP_GT, count, disColName, disColValues, container, idxMeta);
	            stack.push(cassExpr);
	            return cassExpr;
	        }
	        else if (right instanceof CassandraLiteral && left instanceof CassandraFieldExpression)
	        {
	       	    CassandraFieldExpression fieldExpr = (CassandraFieldExpression)left;
	       	    CassandraLiteral litExpr = (CassandraLiteral)right;
	       	    IndexContainer container = new IndexContainer(idxMeta.getItem(), fieldExpr.getIndexName());
	            CassandraExpression cassExpr = new CassandraBooleanExpression(client, ByteBufferUtil.string(fieldExpr.getColumnName()),
	                litExpr.getValue(), Expression.OP_LT, count, disColName, disColValues, container, idxMeta);
	            stack.push(cassExpr);
	            return cassExpr;
	        }
    	}catch (Exception e){
    		throw new NucleusException(e.getMessage(), e);    		
    	}
    	
    	stack.push(CassandraUnSupportedExpression.getInstance());
    	return null;
    }

    /* (non-Javadoc)
     * @see org.datanucleus.query.evaluator.AbstractExpressionEvaluator#processGteqExpression(org.datanucleus.query.expression.Expression)
     */
    @Override
    protected Object processGteqExpression(Expression expr)
    {       
    	try{
	    	Object right = stack.pop();
	        Object left = stack.pop();
	        if (left instanceof CassandraLiteral && right instanceof CassandraFieldExpression)
	        {
	       	    CassandraFieldExpression fieldExpr = (CassandraFieldExpression)right;
	            CassandraLiteral litExpr = (CassandraLiteral)left;
	            IndexContainer container = new IndexContainer(idxMeta.getItem(), fieldExpr.getIndexName());
	            CassandraExpression cassExpr = new CassandraBooleanExpression(client, ByteBufferUtil.string(fieldExpr.getColumnName()),
	                litExpr.getValue(), Expression.OP_LTEQ, count, disColName, disColValues, container, idxMeta);
	            stack.push(cassExpr);
	            return cassExpr;
	        }
	        else if (right instanceof CassandraLiteral && left instanceof CassandraFieldExpression)
	        {
	       	    CassandraFieldExpression fieldExpr = (CassandraFieldExpression)left;
	       	    CassandraLiteral litExpr = (CassandraLiteral)right;
	       	    IndexContainer container = new IndexContainer(idxMeta.getItem(), fieldExpr.getIndexName());
	            CassandraExpression cassExpr = new CassandraBooleanExpression(client, ByteBufferUtil.string(fieldExpr.getColumnName()),
	                litExpr.getValue(), Expression.OP_GTEQ, count, disColName, disColValues, container, idxMeta);
	            stack.push(cassExpr);
	            return cassExpr;
	        }
    	}catch (Exception e){
    		throw new NucleusException(e.getMessage(), e);    		
    	}
    	
    	stack.push(CassandraUnSupportedExpression.getInstance());
    	return null;
    }

    /* (non-Javadoc)
     * @see org.datanucleus.query.evaluator.AbstractExpressionEvaluator#processLteqExpression(org.datanucleus.query.expression.Expression)
     */
    @Override
    protected Object processLteqExpression(Expression expr)
    {
    	try{
	    	Object right = stack.pop();
	        Object left = stack.pop();
	        if (left instanceof CassandraLiteral && right instanceof CassandraFieldExpression)
	        {
	       	    CassandraFieldExpression fieldExpr = (CassandraFieldExpression)right;
	            CassandraLiteral litExpr = (CassandraLiteral)left;
	            IndexContainer container = new IndexContainer(idxMeta.getItem(), fieldExpr.getIndexName());
	            CassandraExpression cassExpr = new CassandraBooleanExpression(client, ByteBufferUtil.string(fieldExpr.getColumnName()),
	                litExpr.getValue(), Expression.OP_GTEQ, count, disColName, disColValues, container, idxMeta);
	            stack.push(cassExpr);
	            return cassExpr;
	        }
	        else if (right instanceof CassandraLiteral && left instanceof CassandraFieldExpression)
	        {
	       	    CassandraFieldExpression fieldExpr = (CassandraFieldExpression)left;
	       	    CassandraLiteral litExpr = (CassandraLiteral)right;
	       	    IndexContainer container = new IndexContainer(idxMeta.getItem(), fieldExpr.getIndexName());
	            CassandraExpression cassExpr = new CassandraBooleanExpression(client, ByteBufferUtil.string(fieldExpr.getColumnName()),
	                litExpr.getValue(), Expression.OP_LTEQ, count, disColName, disColValues, container, idxMeta);
	            stack.push(cassExpr);
	            return cassExpr;
	        }
    	}catch (Exception e){
    		throw new NucleusException(e.getMessage(), e);    		
    	}
    	
    	stack.push(CassandraUnSupportedExpression.getInstance());
    	return null;
    }

    /* (non-Javadoc)
     * @see org.datanucleus.query.evaluator.AbstractExpressionEvaluator#processPrimaryExpression(org.datanucleus.query.expression.PrimaryExpression)
     */
    @Override
    protected Object processPrimaryExpression(PrimaryExpression expr)
    {
        Expression left = expr.getLeft();
        if (left == null)
        {
            
        	AbstractMemberMetaData member = candidateCmd.getMetaDataForMember(expr
    				.getSymbol().getQualifiedName());

    	    ByteBuffer columnName = MetaDataUtils.getColumnName(candidateCmd,
        				member.getAbsoluteFieldNumber());
    	    
            CassandraFieldExpression fieldExpr = new CassandraFieldExpression(null, columnName);
            String indexName = MetaDataUtils.getIndexName(member);
			if(indexName != null)
				fieldExpr.setIndexName(indexName);
			else
				throw new NucleusException("Query on  no-index column is not supported by this mapper");
            stack.push(fieldExpr);
            return fieldExpr;
            
        }

        // TODO Auto-generated method stub
        stack.push(CassandraUnSupportedExpression.getInstance());
    	return null;
    }

    /* (non-Javadoc)
     * @see org.datanucleus.query.evaluator.AbstractExpressionEvaluator#processParameterExpression(org.datanucleus.query.expression.ParameterExpression)
     */
    @Override
    protected Object processParameterExpression(ParameterExpression expr)
    {
        if (expr.getPosition() >= 0)
        {
            if (paramNameByPosition == null)
            {
                paramNameByPosition = new HashMap<Integer, String>();
            }
            paramNameByPosition.put(Integer.valueOf(expr.getPosition()), expr.getId());
        }

        Object paramValue = null;
        boolean paramValueSet = false;
        if (parameters != null && parameters.size() > 0)
        {
            // Check if the parameter has a value
            if (parameters.containsKey(expr.getId()))
            {
                // Named parameter
                paramValue = parameters.get(expr.getId());
                paramValueSet = true;
            }
            else if (parameterValueByName != null && parameterValueByName.containsKey(expr.getId()))
            {
                // Positional parameter, but already encountered
                paramValue = parameterValueByName.get(expr.getId());
                paramValueSet = true;
            }
            else
            {
                // Positional parameter, not yet encountered
                int position = positionalParamNumber;
                if (positionalParamNumber < 0)
                {
                    position = 0;
                }
                if (parameters.containsKey(Integer.valueOf(position)))
                {
                    paramValue = parameters.get(Integer.valueOf(position));
                    paramValueSet = true;
                    positionalParamNumber = position+1;
                    if (parameterValueByName == null)
                    {
                        parameterValueByName = new HashMap<String, Object>();
                    }
                    parameterValueByName.put(expr.getId(), paramValue);
                }
            }
        }

        if (paramValueSet)
        {
            Object idxValue = converterContext.getIndexValue(paramValue);
        	if (idxValue != null)
            {
                CassandraLiteral paramLit = new CassandraLiteral(idxValue);
                stack.push(paramLit);
                return paramLit;
            }
        }

        // TODO Auto-generated method stub
        stack.push(CassandraUnSupportedExpression.getInstance());
    	return null;
    }

    /* (non-Javadoc)
     * @see org.datanucleus.query.evaluator.AbstractExpressionEvaluator#processLiteral(org.datanucleus.query.expression.Literal)
     */
    @Override
    protected Object processLiteral(Literal expr)
    {
        Object litValue = expr.getLiteral();
        if (litValue instanceof Number || litValue instanceof String )
        {
            CassandraLiteral lit = new CassandraLiteral(litValue);
            stack.push(lit);
            return lit;
        }        
        // TODO Handle all Cassandra supported (literal) types

        stack.push(CassandraUnSupportedExpression.getInstance());
    	return null;
    }
   
}